package com.atguigu.gmall.realtime.app.func;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.beans.TableProcess;
import com.atguigu.gmall.realtime.common.GmallConfig;
import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ReadOnlyBroadcastState;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.*;

/**
 * Author: Felix
 * Date: 2021/11/26
 * Desc: 业务数据动态分流实现类
 */
public class TableProcessFunction extends BroadcastProcessFunction<JSONObject, String, JSONObject> {

    private OutputTag<JSONObject> dimTag;

    private MapStateDescriptor<String, TableProcess> mapStateDescriptor;

    private Connection conn;

    public TableProcessFunction(OutputTag<JSONObject> dimTag, MapStateDescriptor<String, TableProcess> mapStateDescriptor) {
        this.mapStateDescriptor = mapStateDescriptor;
        this.dimTag = dimTag;
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        //注册驱动
        Class.forName("org.apache.phoenix.jdbc.PhoenixDriver");
        //创建连接
        conn = DriverManager.getConnection(GmallConfig.PHOENIX_SERVER);
    }

    //处理主流业务数据
    @Override
    public void processElement(JSONObject jsonObj, ReadOnlyContext ctx, Collector<JSONObject> out) throws Exception {
        //获取状态
        ReadOnlyBroadcastState<String, TableProcess> broadcastState = ctx.getBroadcastState(mapStateDescriptor);

        //获取业务数据库表名
        String tableName = jsonObj.getString("table");
        //获取操作类型
        String type = jsonObj.getString("type");
        //注意：后续我们使用maxwell提供的对历史数据进行处理的命令maxwell-bootstrap对历史数据进行处理的时候，类型为bootstrap-insert
        if ("bootstrap-insert".equals(type)) {
            type = "insert";
            jsonObj.put("type", type);
        }

        //根据表名和操作类型拼接key
        String key = tableName + ":" + type;

        //从状态中获取当前数据的配置信息
        TableProcess tableProcess = broadcastState.get(key);

        if (tableProcess != null) {
            //判断当前处理的业务数据是维度还是事实
            String sinkType = tableProcess.getSinkType();
            //不管是事实还是维度   在向下游传递数据的时候，都需要知道目的地
            jsonObj.put("sink_table", tableProcess.getSinkTable());
            //不管是事实还是维度    这里都做一个字段的过滤
            JSONObject dataJsonObj = jsonObj.getJSONObject("data");
            //对data中的内容进行过滤
            filterColumn(dataJsonObj,tableProcess.getSinkColumns());

            if (TableProcess.SINK_TYPE_HBASE.equals(sinkType)) {
                //维度数据   放到维度侧输出流中
                ctx.output(dimTag, jsonObj);
            } else if (TableProcess.SINK_TYPE_KAFKA.equals(sinkType)) {
                //事实数据  放到主流中
                out.collect(jsonObj);
            }
        } else {
            System.out.println("no this key in TableProcess:" + key);
        }

    }

    /**
     * 字段过滤
     * @param dataJsonObj    {"tm_name":"aa","logo_url":"11","id":15}
     * @param sinkColumns    id,tm_name
     */
    private void filterColumn(JSONObject dataJsonObj, String sinkColumns) {
        //将保留的字段放到集合中  方便进行包含判断
        String[] fieldArr = sinkColumns.split(",");
        List<String> fieldList = Arrays.asList(fieldArr);
        //获取json对象的所有名值对
        Set<Map.Entry<String, Object>> entrySet = dataJsonObj.entrySet();
        entrySet.removeIf(entry->!fieldList.contains(entry.getKey()));
    }

    //处理广播流中的配置数据
    @Override
    public void processBroadcastElement(String jsonStr, Context ctx, Collector<JSONObject> out) throws Exception {
        //jsonStr  是FlinkCDC从MySQL中读取配置信息，然后我们通过自定义反序列化器的方式  将其封装为jsonStr
        /*{"database":"gmall0609_realtime","data":{"operate_type":"insert","sink_type":"hbase",
            "sink_table":"dim_base_trademark","source_table":"base_trademark","sink_pk":"id","sink_columns":"id,tm_name"},
            "type":"insert","table":"table_process"}*/
        //为了处理方便，将json字符串转换为json对象
        JSONObject jsonObj = JSON.parseObject(jsonStr);
        //获取从配置表中读取的一条配置记录--->我们可以将其转换为TableProcess实体类对象
        JSONObject dataJsonObj = jsonObj.getJSONObject("data");
        //将配置记录封装为TableProcess对象
        TableProcess tableProcess = dataJsonObj.toJavaObject(TableProcess.class);
        //获取原始数据库表名
        String sourceTable = tableProcess.getSourceTable();
        //获取操作类型
        String operateType = tableProcess.getOperateType();
        //获取是维度还是事实
        String sinkType = tableProcess.getSinkType();
        //获取输出的目的地
        String sinkTable = tableProcess.getSinkTable();
        //获取建表字段
        String sinkColumns = tableProcess.getSinkColumns();
        //获取建表扩展
        String sinkExtend = tableProcess.getSinkExtend();
        //获取主键
        String sinkPk = tableProcess.getSinkPk();

        //根据表名和操作类型 拼接key
        String key = sourceTable + ":" + operateType;

        //对读取到的配置信息进行判断     如果是维度配置，并且配置的是insert操作，那么提前创建维度表
        if (TableProcess.SINK_TYPE_HBASE.equals(sinkType) && "insert".equals(operateType)) {
            checkTable(sinkTable, sinkColumns, sinkPk, sinkExtend);
        }


        //获取广播状态
        BroadcastState<String, TableProcess> broadcastState = ctx.getBroadcastState(mapStateDescriptor);
        //将配置信息放到状态中保存
        broadcastState.put(key, tableProcess);
    }

    //维度表的创建
    private void checkTable(String tableName, String fields, String pk, String ext) {
        //处理主键或者建表扩展为null的情况
        if (pk == null) {
            pk = "id";
        }
        if (ext == null) {
            ext = "";
        }
        //拼接建表语句
        StringBuilder createSql = new StringBuilder("create table if not exists " + GmallConfig.HBASE_SCHEMA + "." + tableName + "(");

        //建表字段的处理
        String[] fieldArr = fields.split(",");
        for (int i = 0; i < fieldArr.length; i++) {
            String field = fieldArr[i];
            if (i >= 1) {
                createSql.append(",");
            }
            if (field.equals(pk)) {
                createSql.append(field + " varchar primary key ");
            } else {
                createSql.append(field + " varchar ");
            }

           /* //如果是最后一个字段，那不需要加逗号
            if(i < fieldArr.length -1){
                createSql.append(",");
            }*/
        }

        createSql.append(")" + ext);

        System.out.println("在phoenix中建表的语句为:" + createSql);

        PreparedStatement ps = null;
        try {
            //获取数据库操作对象
            ps = conn.prepareStatement(createSql.toString());
            //执行SQL语句
            ps.execute();
        } catch (Exception e) {
            e.printStackTrace();
            throw new RuntimeException("在phoenix中创建维度表失败");
        }finally {
            //释放资源
            if(ps != null){
                try {
                    ps.close();
                } catch (SQLException e) {
                    e.printStackTrace();
                }
            }
        }
    }

}
